package com.peng.my_rocketmq.consumer;

import java.util.concurrent.ConcurrentHashMap;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import com.peng.my_rocketmq.bean.User;

@Component

@RocketMQTransactionListener(txProducerGroup ="my-spring-rocketmq-transa-producer")
public class TransMessageExtProducer implements RocketMQLocalTransactionListener {   
	protected final Logger log = LoggerFactory.getLogger(this.getClass());

	 
    private ConcurrentHashMap<String, Integer> countHashMap = new ConcurrentHashMap<>();
    
    private final static int MAX_COUNT = 5;
    // 实现消息的消费处理

	@Override
	public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
		  // 
       // String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
        // 将bizUniNo入库，表名：t_message_transaction,表结构  bizUniNo(主键),业务类型。
       // return LocalTransactionState.UNKNOW;
		
		
		
		
		     // 实现执行本地事务的逻辑
		    	System.out.println("execute---------------------- " +msg.toString());
		    	
		    	int value =Integer.valueOf(msg.getHeaders().get("rocketmq_KEYS").toString().substring(4));
		      // 可以根据具体的本地事务的执行情况返回 RocketMQLocalTransactionState.COMMIT, ROLLBACK 或UNKNOWN 三种状态
		    	
		    	       log.info("value{},余{} body的内容{}", value,value%3,new String(msg.toString()));        
		    	        if ((value % 3) == 0) {
		    	            return RocketMQLocalTransactionState.COMMIT;
		    	        } else if ((value % 3) == 1) {
		    	            return RocketMQLocalTransactionState.ROLLBACK;
		    	        } else{
		    	            return RocketMQLocalTransactionState.UNKNOWN;
		    	        }
	}

	@Override
	public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
		 Integer status = 0;
	        // 从数据库查查询t_message_transaction表，如果该表中存在记录，则提交，
	     //   String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
	        // 然后t_message_transaction 表，是否存在bizUniNo，如果存在，则返回COMMIT_MESSAGE，
	        // 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE
	        
	      //  if(query(bizUniNo) > 0 ) {
	      //      return LocalTransactionState.ROLLBACK_MESSAGE;
	      //  }
	        
	    //    return rollBackOrUnown(bizUniNo);
		
		
		
		// 实现本地事务回查的逻辑
	     // 可以根据具体的本地事务的执行情况返回 RocketMQLocalTransactionState.COMMIT, ROLLBACK 或UNKNOWN 三种状态      
	    	  System.out.println("checkLocal---------------"+msg.getHeaders().toString());
 	    	  int value =Integer.valueOf(msg.getHeaders().get("rocketmq_KEYS").toString().substring(4));
	    	  if ((value % 3) ==0||(value % 3) ==2) {
	    		  return RocketMQLocalTransactionState.COMMIT;
 	          } else{
 	            return RocketMQLocalTransactionState.ROLLBACK;
 	        }
	  
	}   


	
	public int query(String bizUniNo) {
        return 1; //select count(1) from t_message_transaction a where a.biz_uni_no=#{bizUniNo}
    }
    
    public LocalTransactionState rollBackOrUnown(String bizUniNo) {
        Integer num = countHashMap.get(bizUniNo);
        
        if(num != null &&  ++num > MAX_COUNT) {
            countHashMap.remove(bizUniNo);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        
        if(num == null) {
            num = new Integer(1);
        }
        
        countHashMap.put(bizUniNo, num);
        return LocalTransactionState.UNKNOW;
        
    }
	
	
}

